package com.example.pushMessage;

import com.example.entity.Request;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Component;

import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.Arrays;
import java.util.Collection;
import java.util.concurrent.ConcurrentHashMap;

@Component
public class RocketmqHandlerRegistry {
    private final Logger logger = LoggerFactory.getLogger(RocketmqHandlerRegistry.class);

    private volatile boolean initial;
    private Environment environment;
    private ApplicationContext applicationContext;

    private ConcurrentHashMap<String, RocketmqMessageHandler> handlers;
    private ConcurrentHashMap<String, ParameterizedType> handlerParameters;

    public RocketmqHandlerRegistry(Environment environment,
                                   ApplicationContext applicationContext) {
        this.environment = environment;
        this.applicationContext = applicationContext;

        handlers = new ConcurrentHashMap<>();
        handlerParameters = new ConcurrentHashMap<>();

        initialization();
    }

    private void initialization() {
        Collection<RocketmqMessageHandler> candidates =
                applicationContext.getBeansOfType(RocketmqMessageHandler.class).values();
        if (candidates.size() == 0) return;

        candidates.forEach(handler -> {

            Class<?> clazz = null;
            try {
                Class current = handler.getClass();
                do {
                    Type[] types = current.getGenericInterfaces();
                    if (types.length == 0) {
                        current = current.getSuperclass();
                    } else {
                        clazz = (Class<?>) ((ParameterizedType) types[0]).getActualTypeArguments()[1];
                        break;
                    }
                } while (current != Object.class);
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
            }
            if (clazz == null) return;

            String topic = environment.resolvePlaceholders(handler.topic(clazz));
            String tag = environment.resolvePlaceholders(handler.selectorExpression());
            String destination = String.format("%s:%s", topic, tag);
            handlers.put(destination, handler);

            /**
             * for some reason to fetch parameter Type of method
             * it is from java generic type erase
             */
            ParameterizedType requestType =
                    Arrays.stream(handler.getClass().getMethods())
                            .filter(method -> {
                                if (!method.getName().equals("consuming")) {
                                    return false;
                                } else {
                                    Class<?> type = method.getParameterTypes()[0];
                                    if (type == Object.class) return false;
                                    return Request.class.isAssignableFrom(type);
                                }
                            })
                            .map(m -> (ParameterizedType) m.getGenericParameterTypes()[0])
                            .findFirst()
                            .orElse(null);

            if (requestType == null) {
                throw new RuntimeException(String.format("%s declaration is incorrect", handler.getClass().toString()));
            }

            handlerParameters.putIfAbsent(destination, requestType);
        });

        initial = true;
    }

    boolean initialized() {
        return initial;
    }

     public RocketmqMessageHandler getHandler(String destination) {
        if (!handlers.containsKey(destination)) {
            logger.warn("destination {} do not supported or not found", destination);
            return null;
        } else {
            return handlers.get(destination);
        }
    }

     public ParameterizedType getRequestType(String destination) {
        return handlerParameters.get(destination);
    }
}